package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.constant.KlineType;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.MinStockWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.map.StockKlineMap;
import cn.itcast.sink.SinkMysql;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

//个股K线
public class StockKlineTask implements ProcessDataInterface {

    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 1.新建侧边流分支（周、月）
         * 2.数据分组
         * 3.划分时间窗口
         * 4.数据处理
         * 5.分流、封装侧边流数据
         * 6.编写插入sql
         * 7.（日、周、月）K线数据写入
         *      数据转换
         *      数据分组
         *      数据写入mysql
         */
        //todo 1.新建侧边流分支（周、月）
        OutputTag<StockBean> weekOpt = new OutputTag<>("weekOpt", TypeInformation.of(StockBean.class));
        OutputTag<StockBean> monthOpt = new OutputTag<>("monthOpt", TypeInformation.of(StockBean.class));
        //todo 2.数据分组
        SingleOutputStreamOperator<StockBean> processData = waterData.keyBy(new KeyFunction())
                .timeWindow(Time.seconds(60))
                .apply(new MinStockWindowFunction())
                .process(new ProcessFunction<StockBean, StockBean>() {
                    @Override
                    public void processElement(StockBean stockBean, ProcessFunction<StockBean, StockBean>.Context context, Collector<StockBean> collector) throws Exception {
                        //日K
                        collector.collect(stockBean);
                        //周K
                        context.output(weekOpt, stockBean);
                        //周K
                        context.output(monthOpt, stockBean);
                    }
                });
        //todo 3.划分时间窗口
        //todo 4.数据处理
        //todo 5.分流、封装侧边流数据
        //todo 6.编写插入sql
        String sql = "replace into %s values(?,?,?,?,?,?,?,?,?,?,?,?,?)";
        //todo 7.（日、周、月）K线数据写入
        //     数据转换 stock->row
        //     数据分组
        //     数据写入mysql
        processData.map(new StockKlineMap(KlineType.DAYK.getType(),KlineType.DAYK.getFirstTxDateType()))
                //提升并行度，提高写入效率
                .keyBy(new KeySelector<Row, String>() {
                            @Override
                            public String getKey(Row row) throws Exception {
                                return row.getField(2).toString();
                            }
                        }).addSink(new SinkMysql(String.format(sql, QuotConfig.config.getProperty("mysql.stock.sql.day.table"))))
                        .setParallelism(20);

        processData.getSideOutput(weekOpt).map(new StockKlineMap(KlineType.WEEKK.getType(),KlineType.WEEKK.getFirstTxDateType()))
                        .keyBy(new KeySelector<Row, String>() {
                            @Override
                            public String getKey(Row row) throws Exception {
                                return row.getField(2).toString();
                            }
                        }).addSink(new SinkMysql(String.format(sql,QuotConfig.config.getProperty("mysql.stock.sql.week.table"))))
                        .setParallelism(20);

        processData.getSideOutput(monthOpt).map(new StockKlineMap(KlineType.MONTHK.getType(),KlineType.MONTHK.getFirstTxDateType()))
                .keyBy(new KeySelector<Row, String>() {
                    @Override
                    public String getKey(Row row) throws Exception {
                        return row.getField(2).toString();
                    }
                }).addSink(new SinkMysql(String.format(sql,QuotConfig.config.getProperty("mysql.stock.sql.month.table"))))
                .setParallelism(20);

    }
}
